package com.pulse.base.service.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.pulse.base.domain.*;
import com.pulse.base.service.*;
import com.pulse.base.utils.ByteBufferMultipartFile;
import com.pulse.common.core.redis.RedisCache;
import com.pulse.common.utils.DateUtils;
import com.pulse.common.utils.StringToMultipartFile;
import com.pulse.common.utils.file.FileUploadUtils;
import com.pulse.common.utils.sor.Sor;
import com.pulse.common.utils.sor.domain.SorInfo;
import com.pulse.common.utils.spring.SpringUtils;
import com.pulse.framework.web.domain.server.SysFile;
import com.pulse.base.domain.vo.DeviceInfo;
import com.pulse.mqtt.MqttConsumer;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.multipart.MultipartFile;

import java.nio.ByteBuffer;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName CtrlMsgThread
 * @Description
 * @Author 陆帅
 * @Date 2024/7/16 13:31
 */
@Component
public class CtrlMsgThread implements Runnable{

    public final static Logger log = LoggerFactory.getLogger(CtrlMsgThread.class);

    private IDeviceService deviceService = (IDeviceService) SpringUtils.getBean(DeviceServiceImpl.class);
    private RedisCache redisCache = (RedisCache) SpringUtils.getBean(RedisCache.class);
    private ICircuitMeasureService circuitMeasureService = (ICircuitMeasureService) SpringUtils.getBean(CircuitMeasureServiceImpl.class);
    private IFilePathService filePathService = (IFilePathService) SpringUtils.getBean(FilePathServiceImpl.class);
    private ICircuitMeasureEventService circuitMeasureEventService = (ICircuitMeasureEventService) SpringUtils.getBean(CircuitMeasureEventServiceImpl.class);
    private ICircuitMeasureReferService circuitMeasureReferService = (ICircuitMeasureReferService) SpringUtils.getBean(CircuitMeasureReferServiceImpl.class);
    private ICircuitThresholdService circuitThresholdService = (ICircuitThresholdService) SpringUtils.getBean(CircuitThresholdServiceImpl.class);
    private IAlarmService alarmService = (IAlarmService) SpringUtils.getBean(AlarmServiceImpl.class);


    private String[] topicArr;

    /**
     * mqtt订阅主题
     */
    private String topic;
    /**
     * 设备SN
     */
    private String deviceSn;

    /**
     * 消息
     */
    private MqttMessage message;

    public CtrlMsgThread(){}

    public CtrlMsgThread(String[] topicArr, String topic, MqttMessage message){
        this.topicArr = topicArr;
        this.topic = topic;
        this.deviceSn = topicArr[2];
        this.message = message;
    }

    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see Thread#run()
     */
    @Override
    public void run() {
        try{
            JSONObject object = JSONObject.parseObject(message.toString());
            String cmd = object.getString("cmd");
            switch (cmd){
                case "get_dev_info":
                    handlerDeviceInfo(object);
                    break;
                case "start_otdr_measure":
                    handlerStartOtdrMeasure(object);
                    break;
                case "report_wd&el":
                    handlerWdEl(object);
                    break;
                default:
                    break;
            }
        }
        catch (Exception e){
            log.error("异常：",e);
            log.error("处理上报数据异常：" + topic);
        }
    }

    /**
     * 解析设备信息 更新数据库设备信息的软硬件版本
     * @param data
     */
    private void handlerDeviceInfo(JSONObject data){
        String deviceNo = data.getString("from");
        JSONObject response = data.getJSONObject("response");
        if(response == null) return;

        Device device = deviceService.selectDeviceByDeviceNo(deviceNo);
        if(device == null) return;

        // 设置设备通道数
        Long num = 1L;
        JSONObject osw = response.getJSONObject("osw");
        if(osw != null){
            Long channelNum = osw.getLongValue("port_cnt");
            if(channelNum != null){
                num = channelNum;
            }
        }

        if(num.equals(0L)){
            num = 1L;
        }

        if(!device.getChannelNum().equals(num)){
            redisCache.push("DeviceChannelNum",deviceNo+":"+num);
        }

        DeviceInfo deviceInfo = response.getObject("dev_info",DeviceInfo.class);
        if(deviceInfo == null) return;

        device.setDeviceModel(deviceInfo.getEquipment_model());
        device.setHardwareVersion(deviceInfo.getHardware_version());
        device.setSoftwareVersion(deviceInfo.getSoftware_version());
        deviceService.updateDevice(device);
    }

    /**
     * 开始测试消息 启动测试是否正常
     * @param data
     */
    private void handlerStartOtdrMeasure(JSONObject data){
        String deviceNo = data.getString("from");
        JSONObject response = data.getJSONObject("response");
        if (response == null) return;

        int startResult = response.getIntValue("start_result");
        if (startResult == 1) {
            return;
        }

        Integer channel = response.getInteger("osw_port_id");
        if (channel == null) return;
        if (channel.equals(0)) {
            channel = 1;
        }

        // 获取光路监测表
        CircuitMeasure circuitMeasure = redisCache.getCacheObject("CircuitMeasureBack:" + deviceNo + "_" + channel);
        if (circuitMeasure == null) {
            return;
        }

        switch (startResult){
            case 0:
                circuitMeasure.setRemark("模块繁忙");
                break;
            case -1:
                circuitMeasure.setRemark("OT故障");
                break;
            case -2:
                circuitMeasure.setRemark("OSW故障");
                break;
            case -3:
                circuitMeasure.setRemark("OSW无效");
                break;
            case -4:
                circuitMeasure.setRemark("OT消失");
            default:
                break;
        }
        circuitMeasure.setCircuitStatus(1);
        circuitMeasure.setStatus(11);
        circuitMeasureService.updateCircuitMeasure(circuitMeasure);

        // 任务完成
        redisCache.setCacheObject("JobStatus:"+deviceNo,"0",10, TimeUnit.MINUTES);
    }

    /**
     * 测试结束 接收测试结果
     * @param data
     */
    private void handlerWdEl(JSONObject data){
        String deviceNo = data.getString("from");
        Integer channel = data.getJSONObject("report").getIntValue("osw_port_id");
        if(channel.equals(0)){
            channel = 1;
        }

        // 获取曲线
        CircuitMeasure circuitMeasure = redisCache.getCacheObject("CircuitMeasureBack:" + deviceNo+"_"+channel);
        if(circuitMeasure == null){
            return;
        }

        Integer testResult = data.getJSONObject("report").getIntValue("test_result");
        if(testResult<0){ // 设备测试异常
            circuitMeasure.setCircuitStatus(0);
            circuitMeasure.setStatus(11);
            circuitMeasure.setRemark("设备端测试异常,错误码：" + testResult);
            circuitMeasureService.updateCircuitMeasure(circuitMeasure);
            redisCache.setCacheObject("JobStatus:"+deviceNo,"0",10, TimeUnit.MINUTES);
            return;
        }else if(testResult==0){  // 用户主动停止
            circuitMeasure.setCircuitStatus(0);
            circuitMeasure.setStatus(11);
            circuitMeasure.setRemark("用户主动停止测试");
            circuitMeasureService.updateCircuitMeasure(circuitMeasure);
            redisCache.setCacheObject("JobStatus:"+deviceNo,"0",10, TimeUnit.MINUTES);
            return;
        }

        // 获取波形
        JSONObject wd = data.getJSONObject("report");

        long startTime = System.currentTimeMillis();

        String fileUrl = null;
        Date dateTime = DateUtils.getNowDate();
        String fileName = "getdwel_" +DateUtils.parseDateToStr("yyyyMMddHHmmss",dateTime)+".sor";
        try{
            // 组装SorInfo
            SorInfo info = Sor.CreateSorInfo(null,fileName,deviceNo,channel,dateTime, wd);

            // sor数据
            ByteBuffer sorFileBytes = Sor.make(info);

            // 创建MultipartFile
            MultipartFile multipartFile = new ByteBufferMultipartFile(
                    "file",        // 表单字段名
                    fileName,             // 原始文件名
                    "application/octet-stream",           // 内容类型
                    sorFileBytes             // ByteBuffer数据
            );

            fileUrl = FileUploadUtils.upload(multipartFile);
        }
        catch (Exception e){
            log.error("sor文件保存失败");
        }

        long endTime = System.currentTimeMillis();
        long duration = endTime - startTime;
        System.out.println("代码运行时间: " + duration + " 毫秒");
//
//
//        String fileUrl = null;
//        String fileName = "getdwel_" + DateUtils.dateTimeNow()+".txt";
//        try{
//            String resStr = wd.toJSONString();
//            MultipartFile multipartFile = StringToMultipartFile.stringToMultipartFile(resStr, fileName, "text/plain");
//            fileUrl = FileUploadUtils.upload(multipartFile);
//        }
//        catch (Exception e){
//            log.error("文件上传失败");
//        }

        if(fileUrl == null){
            log.error("文件上传失败");
            circuitMeasure.setStatus(11);
        }
        else{
            FilePath filePath = new FilePath();
            filePath.setFilePath(fileUrl);
            filePath.setFileName(fileName);
            filePathService.insertFilePath(filePath);

            circuitMeasure.setFileId(filePath.getFileId());
            circuitMeasure.setStatus(0);
        }


        circuitMeasure.setDefaultMeasure(0);

        JSONObject response = data.getJSONObject("report");

        circuitMeasure.setParamTestTime(response.getLongValue("testTime",0));
        circuitMeasure.setParamWaveLength(response.getLongValue("waveLength",0));
        circuitMeasure.setParamRd(response.getLongValue("RD",0));
        circuitMeasure.setParamPw(response.getLongValue("PW",0));
        circuitMeasure.setParamIor(response.getString("IOR"));
        circuitMeasure.setParamBackScatter(response.getString("backScatter"));
        circuitMeasure.setParamFinalThreshold(response.getString("finalThreshold"));
        circuitMeasure.setParamReflectThreshold(response.getString("reflectThreshold"));
        circuitMeasure.setParamNonReflectThreshold(response.getString("nonReflectThreshold"));

        // 总回波损耗
        Double totalReturnLoss = response.getDouble("totallReturnLoss");
        // 全程衰耗
        Double wholeAtten = response.getDouble("wholeAtten");
        // 全程均耗
        Double wholeAvgAtten = response.getDouble("wholeAvgAtten");
        // 全程长度
        Double wholeLen = response.getDouble("wholeLen");

        circuitMeasure.setTotalReturnLoss(totalReturnLoss.toString());
        circuitMeasure.setWholeAtten(wholeAtten.toString());
        circuitMeasure.setWholeAvgAtten(wholeAvgAtten.toString());
        circuitMeasure.setWholeLen(wholeLen.toString());

        circuitMeasure.setCircuitStatus(0);


        circuitMeasure.setMeasureMode(1);

        circuitMeasureService.updateCircuitMeasure(circuitMeasure);

        int eventCount = response.getIntValue("event_cnt",0);
        for (int i = 0;i<eventCount;i++){
            CircuitMeasureEvent event = response.getObject("event"+i,CircuitMeasureEvent.class);
            if(event == null) continue;

            event.setMeasureId(circuitMeasure.getMeasureId());

            circuitMeasureEventService.insertCircuitMeasureEvent(event);
        }

        //任务完成标识为0  完成
        redisCache.setCacheObject("JobStatus:"+deviceNo,"0",10, TimeUnit.MINUTES);

        alarmMonitoring(circuitMeasure);
    }


    private void alarmMonitoring(CircuitMeasure circuitMeasure){
        // 获取参考曲线关联信息
        CircuitMeasureRefer circuitMeasureRefer = circuitMeasureReferService.selectCircuitMeasureReferByDeviceNoAndChannel(circuitMeasure.getDeviceNo(), circuitMeasure.getChannel());
        if(circuitMeasureRefer == null){
            // 如果是点名测试，无参考曲线不判断异常
            if(!circuitMeasure.getType().equals(1)){
                circuitMeasure.setCircuitStatus(1);
                circuitMeasure.setStatus(11);
                circuitMeasure.setRemark("无参考曲线");
                circuitMeasureService.updateCircuitMeasure(circuitMeasure);
            }
            return;
        }

        // 获取参考曲线
        CircuitMeasure refer = circuitMeasureService.selectCircuitMeasureByMeasureId(circuitMeasureRefer.getMeasureId());
        if(refer == null){
            // 如果是点名测试，无参考曲线不判断异常
            if(!circuitMeasure.getType().equals(1)){
                circuitMeasure.setCircuitStatus(1);
                circuitMeasure.setStatus(11);
                circuitMeasure.setRemark("无参考曲线");
                circuitMeasureService.updateCircuitMeasure(circuitMeasure);
            }
            return;
        }

        // 获取设置的报警阈值
        CircuitThreshold circuitThreshold = circuitThresholdService.selectThresholdByDeviceNoAndChannel(circuitMeasure.getDeviceNo(), circuitMeasure.getChannel());
        if(circuitThreshold == null){
            // 如果是点名测试，未设置阈值不判断异常
            if(!circuitMeasure.getType().equals(1)){
                circuitMeasure.setCircuitStatus(1);
                circuitMeasure.setStatus(11);
                circuitMeasure.setRemark("无参考曲线");
                circuitMeasureService.updateCircuitMeasure(circuitMeasure);
            }
            return;
        }

        // 全程长度阈值
        Double fullLengthThreshold = circuitThreshold.getFullLengthThreshold();
        // 全程衰耗阈值
        Double fullAttenThreshold = circuitThreshold.getFullAttenThreshold();

        // 参考全程长度
        Double fullLengthRefer = circuitMeasureRefer.getFullLength() * 1000; // 全程长度 单位换算成米
        // 参考全程衰耗
        Double fullAttenRefer = circuitMeasureRefer.getFullAtten();

        try{
            // 当前全程长度
            Double fullLengthCurr = Double.parseDouble(circuitMeasure.getWholeLen())*1000;
            // 当前全程衰耗
            Double fullAttenCurr = Double.parseDouble(circuitMeasure.getWholeAtten());

            // 如果参考长度 - 当前长度  大于阈值 则报警
            if(fullLengthRefer - fullLengthCurr > fullLengthThreshold){
                Date alarmTime = DateUtils.getNowDate();

                // 组装报警消息
                Alarm alarm = new Alarm();
                String description = String.format(
                        "【全程长度报警】编号为[%s]的设备通道%s所监测的光路在最新检测中显示全程长度减少了%.2f千米，超出预设阈值%.2f千米",
                        circuitMeasure.getDeviceNo(),
                        circuitMeasure.getChannel(),
                        (fullLengthRefer - fullLengthCurr) / 1000.0,
                        fullLengthThreshold / 1000.0
                );
                alarm.setDescription(description);
                alarm.setDeviceNo(circuitMeasure.getDeviceNo());
                alarm.setChannel(circuitMeasure.getChannel());
                alarm.setMeasureId(circuitMeasure.getMeasureId());
                alarm.setReferId(circuitMeasureRefer.getReferId());
                alarm.setAlarmType(3);
                alarm.setLevel(1);
                alarm.setCreateBy("scada");
                alarm.setCreateTime(alarmTime);
                alarm.setType("full_length");
                alarmService.insertAlarm(alarm);

                // 发布报警消息
                String jsonString = JSON.toJSONString(alarm);
                MqttConsumer.publish("$FAMS/OTU/"+circuitMeasure.getDeviceNo()+"/alarm",jsonString);

                // 更新曲线信息
                circuitMeasure.setCircuitStatus(1);
                circuitMeasure.setStatus(0);
                circuitMeasure.setRemark(description);
                circuitMeasureService.updateCircuitMeasure(circuitMeasure);
            }

            // 全程衰耗报警
            if(fullAttenRefer - fullAttenCurr > fullAttenThreshold){
                Date alarmTime = DateUtils.getNowDate();

                // 组装报警消息
                Alarm alarm = new Alarm();
                String description = String.format(
                        "【全程衰耗报警】编号为[%s]的设备通道%s所监测的光路在最新检测中显示全程衰耗增加了%.2fdB，超出预设阈值%.2fdB",
                        circuitMeasure.getDeviceNo(),
                        circuitMeasure.getChannel(),
                        fullAttenRefer - fullAttenCurr,
                        fullAttenThreshold
                );
                alarm.setDescription(description);
                alarm.setDeviceNo(circuitMeasure.getDeviceNo());
                alarm.setChannel(circuitMeasure.getChannel());
                alarm.setMeasureId(circuitMeasure.getMeasureId());
                alarm.setReferId(circuitMeasureRefer.getReferId());
                alarm.setAlarmType(3);
                alarm.setLevel(1);
                alarm.setCreateBy("scada");
                alarm.setCreateTime(alarmTime);
                alarm.setType("full_atten");
                alarmService.insertAlarm(alarm);

                // 发布报警消息
                String jsonString = JSON.toJSONString(alarm);
                MqttConsumer.publish("$FAMS/OTU/"+circuitMeasure.getDeviceNo()+"/alarm",jsonString);

                // 更新曲线信息
                circuitMeasure.setCircuitStatus(1);
                circuitMeasure.setStatus(0);
                circuitMeasure.setRemark(description);
                circuitMeasureService.updateCircuitMeasure(circuitMeasure);
            }
        }
        catch (Exception e){

        }
    }
}
